package cn.iocoder.yudao.module.liveprice.api;

import cn.iocoder.yudao.framework.common.enums.TradeModuleType;
import cn.iocoder.yudao.module.liveprice.api.dto.LivePriceRespDTO;
import cn.iocoder.yudao.module.system.api.price.LivePriceConfApi;
import cn.iocoder.yudao.module.system.api.price.LivePriceConfRespDTO;
import com.alibaba.fastjson.JSONObject;
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.validation.annotation.Validated;

import java.math.BigDecimal;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Service
@Validated
@Slf4j
public class LivePriceApiImpl implements LivePriceApi {

    // Redis中金属价格数据的Hash key
    private static final String METAL_PRICE_HASH_KEY = "live_price:hash";
    // Redis中金属价格版本的Hash key
    private static final String METAL_PRICE_VERSION_HASH_KEY = "live_price:version:hash";
    
    // 缓存过期时间（秒）
    private static final long CACHE_EXPIRATION_SECONDS = 300; // 5分钟
    
    // 本地缓存配置
    private static final int LOCAL_CACHE_SIZE = 1000;
    private static final long LOCAL_CACHE_EXPIRE_SECONDS = 1; // 1秒过期
    
    // 本地缓存
    private final Cache<Integer, LivePriceRespDTO> localCache;
    
    @Resource
    private StringRedisTemplate stringRedisTemplate;
    @Resource
    private LivePriceConfApi livePriceConfApi;

    // Lua脚本：批量更新价格数据和版本号
    private static final String BATCH_UPDATE_PRICE_SCRIPT = 
        "local version_hash = KEYS[1] " +
        "local price_hash = KEYS[2] " +
        "local updates = ARGV[1] " + // JSON数组，包含所有更新数据
        "local current_time = ARGV[2] " +
        "local result = 0 " +
        "local updates_array = cjson.decode(updates) " +
        "for _, update in ipairs(updates_array) do " +
        "    local metal_type = update.metal_type " +
        "    local price_data = update.price_data " +
        "    local old_version = redis.call('HGET', version_hash, metal_type) " +
        "    if not old_version or tonumber(current_time) > tonumber(old_version) then " +
        "        redis.call('HSET', version_hash, metal_type, current_time) " +
        "        redis.call('HSET', price_hash, metal_type, price_data) " +
        "        result = result + 1 " +
        "    end " +
        "end " +
        "redis.call('EXPIRE', version_hash, 3600) " +
        "redis.call('EXPIRE', price_hash, 300) " +
        "return result";

    private final DefaultRedisScript<Long> batchUpdatePriceScript;

    public LivePriceApiImpl() {
        // 初始化本地缓存
        localCache = Caffeine.newBuilder()
                .maximumSize(LOCAL_CACHE_SIZE)
                .expireAfterWrite(LOCAL_CACHE_EXPIRE_SECONDS, TimeUnit.SECONDS)
                .build();
                
        // 初始化Lua脚本
        batchUpdatePriceScript = new DefaultRedisScript<>();
        batchUpdatePriceScript.setScriptText(BATCH_UPDATE_PRICE_SCRIPT);
        batchUpdatePriceScript.setResultType(Long.class);
    }

    @Override
    public List<LivePriceRespDTO> getLivePriceList() {
        return null;
    }

    @Override
    public LivePriceRespDTO getLivePriceByType() {
        return null;
    }

    @Override
    public List<LivePriceRespDTO> getHistory() {
        return null;
    }

    @Override
    @Async("taskExecutor")
    public void asyncCachePriceData(LivePriceRespDTO priceRespDTO) {
        // 将单个更新转换为批量更新
        List<LivePriceRespDTO> updates = Collections.singletonList(priceRespDTO);
        asyncBatchCachePriceData(updates);
    }

    /**
     * 批量异步更新价格数据
     * @param priceRespDTOList 价格数据列表
     */
    @Async("taskExecutor")
    public void asyncBatchCachePriceData(List<LivePriceRespDTO> priceRespDTOList) {
        if (priceRespDTOList == null || priceRespDTOList.isEmpty()) {
            return;
        }

        try {
            // 生成当前时间戳作为版本号
            String currentTime = String.valueOf(System.currentTimeMillis());
            
            // 构建批量更新数据
            List<Map<String, String>> updates = priceRespDTOList.stream()
                    .filter(dto -> dto.getMetalType() != null)
                    .map(dto -> {
                        Map<String, String> update = new HashMap<>();
                        update.put("metal_type", dto.getMetalType().toString());
                        update.put("price_data", JSONObject.toJSONString(dto));
                        return update;
                    })
                    .collect(Collectors.toList());

            if (updates.isEmpty()) {
                return;
            }

            // 使用Lua脚本批量更新
            String updatesJson = JSONObject.toJSONString(updates);
            Long result = stringRedisTemplate.execute(batchUpdatePriceScript,
                    Arrays.asList(METAL_PRICE_VERSION_HASH_KEY, METAL_PRICE_HASH_KEY),
                    updatesJson, currentTime);

            if (result != null && result > 0) {
                // 更新本地缓存
                priceRespDTOList.forEach(dto -> {
                    if (dto.getMetalType() != null) {
                        localCache.put(dto.getMetalType(), dto);
                    }
                });
                log.debug("成功批量更新金属价格数据, 更新数量: {}", result);
            }
        } catch (Exception e) {
            log.error("批量更新金属价格缓存异常", e);
        }
    }

    @Override
    public LivePriceRespDTO getMetalPriceFromCache(Integer metalType) {
        if (metalType == null) {
            return null;
        }

        try {
            // 先查本地缓存
            LivePriceRespDTO cachedPrice = localCache.getIfPresent(metalType);
            if (cachedPrice != null) {
                return cachedPrice;
            }

            // 本地缓存未命中，查Redis
            Object priceJsonObj = stringRedisTemplate.opsForHash().get(METAL_PRICE_HASH_KEY, metalType.toString());
            if (priceJsonObj == null) {
                return null;
            }

            String priceJson = (String) priceJsonObj;
            LivePriceRespDTO priceRespDTO = JSONObject.parseObject(priceJson, LivePriceRespDTO.class);
            
            // 更新本地缓存
            if (priceRespDTO != null) {
                localCache.put(metalType, priceRespDTO);
            }
            
            return priceRespDTO;
        } catch (Exception e) {
            log.error("获取金属价格缓存数据异常, 金属类型: {}", metalType, e);
            return null;
        }
    }

    @Override
    public Map<Integer, LivePriceRespDTO> batchGetMetalPriceFromCache(List<Integer> metalTypes) {
        Map<Integer, LivePriceRespDTO> result = new HashMap<>();
        if (metalTypes == null || metalTypes.isEmpty()) {
            return result;
        }

        try {
            // 先查本地缓存
            metalTypes.forEach(metalType -> {
                LivePriceRespDTO cachedPrice = localCache.getIfPresent(metalType);
                if (cachedPrice != null) {
                    result.put(metalType, cachedPrice);
                }
            });

            // 找出本地缓存未命中的类型
            List<Integer> missingTypes = metalTypes.stream()
                    .filter(type -> !result.containsKey(type))
                    .collect(Collectors.toList());

            if (!missingTypes.isEmpty()) {
                // 批量获取Redis数据
                List<Object> metalTypeObjects = missingTypes.stream()
                        .map(String::valueOf)
                        .collect(Collectors.toList());
                List<Object> priceJsonList = stringRedisTemplate.opsForHash().multiGet(METAL_PRICE_HASH_KEY, metalTypeObjects);

                // 处理结果
                for (int i = 0; i < missingTypes.size(); i++) {
                    Integer metalType = missingTypes.get(i);
                    Object priceJsonObj = priceJsonList.get(i);

                    if (priceJsonObj != null) {
                        try {
                            String priceJson = (String) priceJsonObj;
                            LivePriceRespDTO priceRespDTO = JSONObject.parseObject(priceJson, LivePriceRespDTO.class);
                            if (priceRespDTO != null) {
                                result.put(metalType, priceRespDTO);
                                // 更新本地缓存
                                localCache.put(metalType, priceRespDTO);
                            }
                        } catch (Exception e) {
                            log.error("解析金属价格缓存数据异常, 金属类型: {}", metalType, e);
                        }
                    }
                }
            }
        } catch (Exception e) {
            log.error("批量获取金属价格缓存数据异常", e);
        }

        return result;
    }

    @Override
    public BigDecimal getConfPrice(Integer metalType, boolean sale) {
        LivePriceRespDTO priceRespDTO = getMetalPriceFromCache(metalType);
        if (priceRespDTO == null) {
            return null;
        }
        
        LivePriceConfRespDTO priceConfRespDTO = livePriceConfApi.getConf(TradeModuleType.METAL.getType(), metalType);
        if (priceConfRespDTO == null) {
            return sale ? priceRespDTO.getSalePrice() : priceRespDTO.getReclaimPrice();
        }

        BigDecimal res;
        if (sale) {
            res = priceRespDTO.getSalePrice().add(priceConfRespDTO.getIncreasePrice());
        } else {
            res = priceRespDTO.getReclaimPrice().subtract(priceConfRespDTO.getDecreasePrice());
        }
        return res;
    }
}
